package scalaKafka

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import kafka.utils.{ZKGroupTopicDirs, ZkUtils}
import org.I0Itec.zkclient.ZkClient
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Duration, StreamingContext}

object DirectConnectKafka {
    def main(args: Array[String]): Unit = {
        //指定组
        val group = "g001"
        //创建SparkConf
        val conf = new SparkConf().setAppName("KafkaDirectConnect").setMaster("local[*]")
        //创建SparkStreaming，并设置间隔时间
        val ssc = new StreamingContext(conf, Duration(5000))
        //指定消费的topic名字
        val topic = "wwcc"
        //指定kafka的broker地址（sparkStream的Task直连到kafak的分区上，用更加底层的API消费，效率更高
        val brokerLsit = "node2:9092,node3:9092,node4:9092"

        //指定zk的地址，后期更新消费的偏移量使用 也可以用redis,mysql来记录偏移量
        val zkQuorum = "node2:2181,node3:2181,node4:2181"
        val topics:Set[String]=Set(topic)
        //创建一个ZKGroupTopicDirs 对象，其实是指定往zk中写入数据的目录，用于保存偏移量
        val topicDirs = new ZKGroupTopicDirs(group, topic)
        //获取zookeeper中的路径"/g001/offsets/wordcount/"
        val zkTopicPath = s"${topicDirs.consumerOffsetDir}"

        //准备kafka中的参数
        val kafkaParams = Map(
            "metadata.broker.list" -> brokerLsit,
            "group.id" -> group,
            //从头开始读取数据
            "auto.offset.rest" -> kafka.api.OffsetRequest.SmallestTimeString
        )

        //zoookeeper的host和ip，创建一个client,用于更新偏移量的
        //是zookeeper的客户端，可以从zk中读取偏移量数据，并更新偏移量
        val zkClient = new ZkClient(zkQuorum)
        //查询该路径下是否子阶段（默认有自己诶按为我们自己保存不同partition时生成的）
        val children = zkClient.countChildren(zkTopicPath)
        var kafkaStream: InputDStream[(String, String)] = null

        //如果zookeeper中有保存offsest，我们会利用这个offset作为kafkaStream的起始位置
        var fromOffsets: Map[TopicAndPartition, Long] = Map()
        //如果保存过offset
        if (children > 0) {
            for (i <- 0 until children) {
                // /goo1/offsets/wordcout/0
                val partitionOffset = zkClient.readData[String](s"$zkTopicPath/${i}")
                val tp = TopicAndPartition(topic, i)
                fromOffsets += (tp -> partitionOffset.toLong)
            }
        //Key: kafka的key values "hello tom hello jerry"
        //这个会将kafka的消息就行transform 最终kafka的数据都会变成 kafka的key,message 这样的tuple
        val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())
        //通过KafkaUtils 创建直连的DStream fromOffsets 参数的作用是 按照前面计算好的偏移量继续消费数据
        //[String,String,StringDecoder,StringDecoder,(String,String)
        //key value       key 的解码方式  value的解码方式
        kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffsets, messageHandler)
         }else{
            //如果未保存，根据 kafkaParam 的配置使用最新(largest)或者最旧的（smallest） offset
            kafkaStream=KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,kafkaParams,topics)//注意待着前缀啊。。。

        }
        //偏移量的范围
        var offsetRanges=Array[OffsetRange]()
        //直连方式只有在KafkaDstream的RDD中才能获取偏移量，那么就不能调用DStream的Transformation
        //只能在kafkaStream调用foreachRDD，获取RDD的偏移量，然后对RDD进行操作
        //依次迭代KafkaDStream中的KafkaRDD
        kafkaStream.foreachRDD{kafkaRDD=>
            //只有kakfkaRDD可以强转成HashOffsetRanges,并获取到偏移量
            offsetRanges=kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
            val lines = kafkaRDD.map(_._2)
            //对RDD进行操作，触发Action
            lines.foreachPartition(partition=>
                partition.foreach(x=>{
                    println(x)
                })
            )
        }
        for(o<-offsetRanges){
            val zkPath=s"${topicDirs.consumerOffsetDir}/${(o.partition)}"
            //将该partition的offset保存到zookeeper
            // /goo1/offsets/wordcout/0/20000
            ZkUtils.updatePersistentPath(zkClient,zkPath,o.untilOffset.toString)
        }

        ssc.start()
        ssc.awaitTermination()
    }

}
